/**
 * Copyright 2019 Inc.
 **/
package com.myz.rabbitmqspring.spring.consume;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.myz.rabbitmqspring.spring.builder.RabbitMqAccessBuilder;
import com.myz.rabbitmqspring.spring.constants.Constants;
import com.myz.rabbitmqspring.spring.dto.DetailResponse;
import com.myz.rabbitmqspring.spring.process.MessageProcess;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 线程池消费端
 *
 * @author maoyz0621 on 19-4-14
 * @version: v1.0
 */
public class ThreadPoolConsumer<T> {

    private static final Logger logger = LoggerFactory.getLogger(ThreadPoolConsumer.class);

    private ExecutorService executor;

    private volatile boolean stop = false;

    private final ThreadPoolConsumerBuilder<T> builder;

    /**
     * Builder模式
     *
     * @param <T>
     */
    public static class ThreadPoolConsumerBuilder<T> {
        private int threadCount;
        private long intervalMils;
        private RabbitMqAccessBuilder mqAccessBuilder;
        private String exchange;
        private String routingKey;
        private String queue;
        private String type = "direct";
        private MessageProcess<T> messageProcess;

        public ThreadPoolConsumerBuilder<T> setThreadCount(int threadCount) {
            this.threadCount = threadCount;
            return this;
        }

        public ThreadPoolConsumerBuilder<T> setIntervalMils(long intervalMils) {
            this.intervalMils = intervalMils;
            return this;
        }

        public ThreadPoolConsumerBuilder<T> setMQAccessBuilder(RabbitMqAccessBuilder mqAccessBuilder) {
            this.mqAccessBuilder = mqAccessBuilder;
            return this;
        }

        public ThreadPoolConsumerBuilder<T> setExchange(String exchange) {
            this.exchange = exchange;
            return this;
        }

        public ThreadPoolConsumerBuilder<T> setRoutingKey(String routingKey) {
            this.routingKey = routingKey;
            return this;
        }

        public ThreadPoolConsumerBuilder<T> setQueue(String queue) {
            this.queue = queue;
            return this;
        }

        public ThreadPoolConsumerBuilder<T> setType(String type) {
            this.type = type;
            return this;
        }

        public ThreadPoolConsumerBuilder<T> setMessageProcess(MessageProcess<T> messageProcess) {
            this.messageProcess = messageProcess;
            return this;
        }

        public ThreadPoolConsumer<T> build() {
            return new ThreadPoolConsumer<T>(this);
        }

    }

    private ThreadPoolConsumer(ThreadPoolConsumerBuilder<T> builder) {
        this.builder = builder;
        /* ThreadPoolExecutor(int corePoolSize,
                        int maximumPoolSize,
                        long keepAliveTime,
                        TimeUnit unit,
                        BlockingQueue<Runnable> workQueue,
                        ThreadFactory threadFactory,
                        RejectedExecutionHandler handler)*/
        executor = new ThreadPoolExecutor(
                builder.threadCount,
                builder.threadCount,
                5,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(3),
                new ThreadFactoryBuilder().setNameFormat("thread_pool_consumer_%d").build(),
                new ThreadPoolExecutor.AbortPolicy());
    }

    /**
     * 1 构造messageConsumer
     * 2 执行consume
     */
    public void start() throws IOException {
        logger.debug("*********** ThreadPoolConsumer start() **************");

        for (int i = 0; i < builder.threadCount; i++) {
            // 创建IMessageConsumer
            final IMessageConsumer messageConsumer = builder.mqAccessBuilder.buildMessageConsumer(builder.exchange,
                    builder.routingKey,
                    builder.queue,
                    builder.messageProcess,
                    builder.type);

            // 执行consume
            executor.execute(() -> {
                while (!stop) {
                    DetailResponse response = messageConsumer.consume();

                    if (builder.intervalMils > 0) {
                        try {
                            TimeUnit.MICROSECONDS.sleep(builder.intervalMils);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }

                    if (!response.isSuccess()) {
                        logger.info("run error " + response.getErrorMessage());
                    }
                }
            });
        }
        // 增加hook监听jvm结束，优雅退出
        Runtime.getRuntime().addShutdownHook(new Thread(this::stop));
    }

    /**
     * 停止
     */
    public void stop() {
        this.stop = true;

        try {
            TimeUnit.MICROSECONDS.sleep(Constants.ONE_SECOND);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }


}
